package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.article.mess.ArticleVisitStreamMess;
import com.heima.model.article.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Configuration
@Slf4j
public class HotArticleStreamHandler {

    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        //创建kstream对象，同时指定从那个topic中接收消息
        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
        stream.map((key,value)->{
            //将消息的key和value进行转换重置
            UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);

            //重置消息的key:1234343434   和  value: likes:1
            return new KeyValue<>(mess.getArticleId().toString(),mess.getType().name()+":"+mess.getAdd());
        })
                //分组统计：按照文章id来进行
                .groupBy((key,value)->key)
                //设定时间窗口
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                //手动设定聚合的value
                .aggregate(new Initializer<String>() {
                    @Override
                    //初始化方法：每个时间窗口内进行聚合之前都会执行一次，只执行一次!
                    public String apply() {
                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                    }
                }, new Aggregator<String, String, String>() {
                    @Override
                    //对stream中每个消息进行聚合处理时都会执行一次该方法
                    /*
                        key: 文章id，例如：1234343434
                        value:  行为数据，例如：likes:1
                        applyValue：初始化方法的返回值："COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0"
                     */
                    public String apply(String key, String value, String applyValue) {
                        //1. 先拿到各项行为的初始值
                        int coll = 0, com = 0, like = 0, view = 0;
                        String[] appArray = applyValue.split(",");
                        for (String app : appArray) {  //app = "COLLECTION:0"
                            String[] split = app.split(":");

                            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                                case COLLECTION:
                                    coll = Integer.parseInt(split[1]);
                                    break;
                                case COMMENT:
                                    com = Integer.parseInt(split[1]);
                                    break;
                                case LIKES:
                                    like = Integer.parseInt(split[1]);
                                    break;
                                case VIEWS:
                                    view = Integer.parseInt(split[1]);
                                    break;
                            }
                        }

                        //2. 针对本次消息行为数据进行累加计算
                        String[] split = value.split(":");  //["likes","1"]
                        switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                            case COLLECTION:
                                coll += Integer.parseInt(split[1]);
                                break;
                            case COMMENT:
                                com += Integer.parseInt(split[1]);
                                break;
                            case LIKES:
                                like += Integer.parseInt(split[1]);
                                break;
                            case VIEWS:
                                view += Integer.parseInt(split[1]);
                                break;
                        }

                        //3. 将本次处理后的结果填充到模板中，作为下次计算的模板使用!
                        String res = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", coll, com, like, view);
                        log.info("本次处理消息的文章id为：{}",key);
                        log.info("本次处理消息的聚合结果为：{}",res);
                        return res;
                    }
                }).toStream()
                .map((key,value)->{
                    return new KeyValue<>(key,formatObj(key,value));
                })
                //将处理好的消息投递给下游topic
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);

        return stream;
    }

    //处理聚合消息的value
    private Object formatObj(Windowed<String> key, String value) {
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setArticleId(Long.valueOf(key.key().toString()));

        String[] appArray = value.split(",");
        for (String app : appArray) {  //app = "COLLECTION:0"
            String[] split = app.split(":");

            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])){
                case COLLECTION:
                    mess.setCollect(Integer.parseInt(split[1]));
                    break;
                case COMMENT:
                    mess.setComment(Integer.parseInt(split[1]));
                    break;
                case LIKES:
                    mess.setLike(Integer.parseInt(split[1]));
                    break;
                case VIEWS:
                    mess.setView(Integer.parseInt(split[1]));
                    break;
            }
        }

        return JSON.toJSONString(mess);
    }
}